package io.kiki.sba.registry.server.data_store.timer;

import io.kiki.sba.registry.common.model.dataserver.Datum;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.NamedThreadFactory;
import io.kiki.sba.registry.util.StringFormatter;
import org.apache.commons.lang.time.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.util.Calendar;
import java.util.Date;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public class CacheDigestTask {
    private static final Logger logger = LoggerFactory.getLogger("CACHE-DIGEST");
    private final ScheduledExecutorService scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("CacheDigestTask"));
    @Autowired
    private DatumStorageDelegate datumStorageDelegate;
    @Autowired
    private ServerConfig serverConfig;

    @PostConstruct
    public boolean init() {
        final int intervalMinutes = serverConfig.getCacheDigestIntervalMinutes();
        if (intervalMinutes <= 0) {
            logger.info("cache digest off with intervalMinutes={}", intervalMinutes);
            return false;
        }
        Date firstDate = new Date();
        firstDate = DateUtils.round(firstDate, Calendar.MINUTE);
        firstDate.setMinutes(firstDate.getMinutes() / intervalMinutes * intervalMinutes + intervalMinutes);
        long firstDelay = firstDate.getTime() - System.currentTimeMillis();
        scheduledThreadPoolExecutor.scheduleAtFixedRate(this::dump, firstDelay, (long) intervalMinutes * 60 * 1000, TimeUnit.MILLISECONDS);
        return true;
    }

    boolean dump() {
        try {
            Map<String, Map<String, Datum>> allMap = datumStorageDelegate.getLocalAll();
            if (!allMap.isEmpty()) {
                for (Entry<String, Map<String, Datum>> entry : allMap.entrySet()) {
                    String dataCenter = entry.getKey();
                    Map<String, Datum> datumMap = entry.getValue();
                    logger.info("size of datum in {} is {}", dataCenter, datumMap.size());
                    for (Entry<String, Datum> dataInfoEntry : datumMap.entrySet()) {
                        String dataInfoId = dataInfoEntry.getKey();
                        Datum data = dataInfoEntry.getValue();
                        Map<String, Publisher> pubMap = data.getRegisterIdToPublisherMap();
                        StringBuilder pubStr = new StringBuilder(4096);
                        if (!CollectionUtils.isEmpty(pubMap)) {
                            for (Publisher publisher : pubMap.values()) {
                                pubStr.append(logPublisher(publisher)).append(";");
                            }
                        }
                        logger.info("[Datum]{},{},{},[{}]", dataInfoId, data.getVersion(), dataCenter, pubStr);
                        // avoid io is busy
                        ConcurrentUtils.sleepUninterruptibly(2, TimeUnit.MILLISECONDS);
                    }
                    int pubCount = datumMap.values().stream().mapToInt(Datum::publisherSize).sum();
                    logger.info("size of publisher in {} is {}", dataCenter, pubCount);
                }
            } else {
                logger.info("datum cache is empty");
            }
            return true;
        } catch (Throwable t) {
            logger.error("cache digest error", (Throwable) t);
        }
        return false;
    }

    private String logPublisher(Publisher publisher) {
        if (publisher != null) {
            Url url = publisher.getClientUrl();
            String urlStr = url != null ? url.buildAddressString() : "null";
            return StringFormatter.format("{},{},{},{}", publisher.getRegisterId(), publisher.getRegisterTimestamp(), urlStr, publisher.getVersion());
        }
        return "";
    }


}
